package com.shujia.util

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object CarUtil {
  /**
    * 读取过车数据的工具
    *
    * @param env ：flink环境
    * @return
    */

  def loadKafkaData(env: StreamExecutionEnvironment): DataStream[String] = {
    /**
      * 实时读取kafka中的数据
      *
      */
    val properties = new Properties()
    //kafka集群地址
    properties.setProperty("bootstrap.servers", "master:9092")
    //消费者组，随便写, 一条消息在一个组内只被消费一次
    properties.setProperty("group.id", "asfasfsafsasdsad")


    //创建kafka消费者
    val consumer = new FlinkKafkaConsumer[String](
      "cars_temp1",
      new SimpleStringSchema(),
      properties)


    //指定读取数据的位置
    consumer.setStartFromEarliest()

    //返回DS
    env.addSource(consumer)
  }
}
